//
//  Publishers.Share
//  
//
//  Created by Sergej Jaskiewicz on 18/09/2019.
//

extension Publisher {

    /// Shares the output of an upstream publisher with multiple subscribers.
    ///
    /// The publisher returned by this operator supports multiple subscribers, all of whom
    /// receive unchanged elements and completion states from the upstream publisher.
    ///
    ///  - Tip: `Publishers.Share` is effectively a combination of
    ///  the `Publishers.Multicast` and `PassthroughSubject` publishers, with an implicit
    ///  `autoconnect()`.
    ///
    /// The following example uses a sequence publisher as a counter to publish three
    /// random numbers, generated by a `map(_:)` operator. It uses a `share()` operator
    /// to share the same random number to each of two subscribers. This example uses
    /// a `delay(for:tolerance:scheduler:options:)` operator only to prevent the first
    /// subscriber from exhausting the sequence publisher immediately; an asynchronous
    /// publisher wouldn’t need this.
    ///
    ///     let pub = (1...3).publisher
    ///         .delay(for: 1, scheduler: DispatchQueue.main)
    ///         .map( { _ in return Int.random(in: 0...100) } )
    ///         .print("Random")
    ///         .share()
    ///
    ///     cancellable1 = pub
    ///         .sink { print ("Stream 1 received: \($0)")}
    ///     cancellable2 = pub
    ///         .sink { print ("Stream 2 received: \($0)")}
    ///
    ///     // Prints:
    ///     // Random: receive value: (20)
    ///     // Stream 1 received: 20
    ///     // Stream 2 received: 20
    ///     // Random: receive value: (85)
    ///     // Stream 1 received: 85
    ///     // Stream 2 received: 85
    ///     // Random: receive value: (98)
    ///     // Stream 1 received: 98
    ///     // Stream 2 received: 98
    ///
    /// Without the `share()` operator, stream 1 receives three random values,
    /// followed by stream 2 receiving three different random values.
    ///
    /// Also note that `Publishers.Share` is a `class` rather than a `struct` like most
    /// other publishers. This means you can use this operator to create a publisher
    /// instance that uses reference semantics.
    ///
    /// - Returns: A class instance that shares elements received from its upstream to
    ///   multiple subscribers.
    public func share() -> Publishers.Share<Self> {
        return .init(upstream: self)
    }
}

extension Publishers {

    /// A publisher that shares the output of an upstream publisher with multiple
    /// subscribers.
    ///
    /// This publisher type supports multiple subscribers, all of whom receive unchanged
    /// elements and completion states from the upstream publisher.
    ///
    ///  - Tip: `Publishers.Share` is effectively a combination of
    ///  the `Publishers.Multicast` and `PassthroughSubject` publishers,
    ///  with an implicit `autoconnect()`.
    ///
    /// Also note that `Publishers.Share` is a `class` rather than a `struct` like most
    /// other publishers. Use this type when you need a publisher instance that uses
    /// reference semantics.
    public final class Share<Upstream: Publisher>: Publisher, Equatable {

        public typealias Output = Upstream.Output

        public typealias Failure = Upstream.Failure

        private typealias MulticastSubject = PassthroughSubject<Output, Failure>

        private let inner: Autoconnect<Multicast<Upstream, MulticastSubject>>

        public let upstream: Upstream

        public init(upstream: Upstream) {
            self.inner = upstream.multicast(subject: .init()).autoconnect()
            self.upstream = upstream
        }

        public func receive<Downstream: Subscriber>(subscriber: Downstream)
            where Downstream.Input == Output, Downstream.Failure == Failure
        {
            inner.subscribe(subscriber)
        }

        public static func == (lhs: Share, rhs: Share) -> Bool {
            return lhs === rhs
        }
    }
}
